package dongshi.daddy.transaction.twopc;

//import dongshi.daddy.util.SocketUtil;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

// 两阶段提交测试代码
public class TwoPcTest {
    private static CountDownLatch allNodeReady = new CountDownLatch(3);
    private static CountDownLatch votePhaseMsgReceived = new CountDownLatch(3);
    private static CountDownLatch receiveCommitPhaseReady = new CountDownLatch(3);
    private static int Coco_PORT = 6666;
    private static int NODE1_PORT = 6667;
    private static int NODE2_PORT = 6668;
    private static int NODE3_PORT = 6669;
    public static void main(String[] args) throws Exception {
        Thread coco = new Thread(() -> {
            try {
                allNodeReady.await();
                System.out.println("所有节点准备完毕，开始进入投票阶段！");
                String votePhaseMsg = "能提交事务吗？";
                sendMsg(votePhaseMsg, NODE1_PORT);
                sendMsg(votePhaseMsg, NODE2_PORT);
                sendMsg(votePhaseMsg, NODE3_PORT);
                votePhaseMsgReceived.await();
                System.out.println("协调器准备接收投票阶段响应！");
                ServerSocket serverNode1 = new ServerSocket(Coco_PORT);

                // 接收投票阶段响应
                new Thread(() -> {
//                    String roundTwoReceivedMsg = null;
                    try {
                        List<String> votePhaseResList = new ArrayList<>();
                        while (votePhaseResList.size() < 3) {
                            String votePhaseRes = receiveMsg(serverNode1);
                            votePhaseResList.add(votePhaseRes);
                        }
                        System.out.println("协调器接收的投票阶段响应是：" + votePhaseResList);
                        receiveCommitPhaseReady.await();
                        System.out.println("协调器发送提交阶段消息开始");
                        String commitPhaseMsg = "提交事务吧";
                        sendMsg(commitPhaseMsg, NODE1_PORT);
                        sendMsg(commitPhaseMsg, NODE2_PORT);
                        sendMsg(commitPhaseMsg, NODE3_PORT);
                        List<String> commitPhaseResList = new ArrayList<>();
                        while (commitPhaseResList.size() < 3) {
                            String commitPhaseRes = receiveMsg(serverNode1);
                            commitPhaseResList.add(commitPhaseRes);
                        }
                        System.out.println("协调器接收的提交阶段响应是：" + commitPhaseResList);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();

            } catch (Exception e) {
            }
        });
        coco.start();

        Thread node1 = new Thread(() -> {
            try {
                ServerSocket serverNode1 = new ServerSocket(NODE1_PORT);
                System.out.println("node1准备好了！");
                allNodeReady.countDown();
                String votePhaseMsg = receiveMsg(serverNode1);
                System.out.println("node1投票阶段收到消息：" + votePhaseMsg);
                votePhaseMsgReceived.countDown();
                votePhaseMsgReceived.await();
                // 发送投票阶段响应
                sendMsg("可以提交", Coco_PORT);
                receiveCommitPhaseReady.countDown();
                String commitPhaseMsg = receiveMsg(serverNode1);
                System.out.println("node1提交阶段收到消息：" + commitPhaseMsg);
                sendMsg("提交完成", Coco_PORT);
            } catch (Exception e) {
            }
        });
        node1.start();

        Thread node2 = new Thread(() -> {
            try {
                ServerSocket serverNode2 = new ServerSocket(NODE2_PORT);
                System.out.println("node2准备好了！");
                allNodeReady.countDown();
                String votePhaseMsg = receiveMsg(serverNode2);
                System.out.println("node2投票阶段收到消息：" + votePhaseMsg);
                votePhaseMsgReceived.countDown();
                votePhaseMsgReceived.await();
                // 发送投票阶段响应
                sendMsg("可以提交", Coco_PORT);
                receiveCommitPhaseReady.countDown();
                String commitPhaseMsg = receiveMsg(serverNode2);
                System.out.println("node1提交阶段收到消息：" + commitPhaseMsg);
                sendMsg("提交完成", Coco_PORT);
            } catch (Exception e) {
            }
        });
        node2.start();

        Thread node3 = new Thread(() -> {
            try {
                ServerSocket serverNode3 = new ServerSocket(NODE3_PORT);
                System.out.println("node3准备好了！");
                allNodeReady.countDown();
                String votePhaseMsg = receiveMsg(serverNode3);
                System.out.println("node3投票阶段收到消息：" + votePhaseMsg);
                votePhaseMsgReceived.countDown();
                votePhaseMsgReceived.await();
                // 发送投票阶段响应
                sendMsg("可以提交", Coco_PORT);
                receiveCommitPhaseReady.countDown();
                String commitPhaseMsg = receiveMsg(serverNode3);
                System.out.println("node1提交阶段收到消息：" + commitPhaseMsg);
                sendMsg("提交完成", Coco_PORT);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        node3.start();

        coco.join();
    }

//    private static void sendMsg(String sendMsg, int port) throws IOException {
//        Socket socket = new Socket("127.0.0.1", port);
//        OutputStream os = socket.getOutputStream();
//        PrintWriter pw = new PrintWriter(os);
//        pw.println(sendMsg);
//        pw.flush();
//        socket.close();
//    }

    public static String receiveMsg(ServerSocket serverA, CountDownLatch countDownLatch) throws IOException {
        countDownLatch.countDown();
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

    public static void sendMsg(String sendMsg, int port) throws Exception {
//        Socket socket = new Socket("127.0.0.1", port);
//        socket.setSoTimeout();
        Socket socket = new Socket();
//        socket.setSoTimeout(30000);
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }
}
